package com.atbeijing.D08;

import org.apache.commons.codec.Charsets;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.shaded.curator4.com.google.common.hash.BloomFilter;
import org.apache.flink.shaded.curator4.com.google.common.hash.Funnels;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;

/**
 * 使用bloom filter(布隆过滤器)去重
 * 求uv数据,即pv数据去重
 *
 * bloom filter 底层是bit类型的数组,每位存0 或1
 * 当字符串保存时,通过字符串和n个hash函数计算出n个位置,将数组中这几个位置由0置为1
 * bloom filter 适用于允许误差的情况下大数据去重,不能保证完全准确(不同字符串计算的hash值可能相同)
 *
 * bloom filter的泛型是他要存放的值的类型
 */
public class Example10 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .readTextFile("C:\\Users\\WangJinGen\\IdeaProjects\\flink\\data\\UserBehavior.csv")
                .flatMap(new FlatMapFunction<String, UserBehavior>() {
                    @Override
                    public void flatMap(String value, Collector<UserBehavior> out) throws Exception {
                        String[] split = value.split(",");
                        out.collect(new UserBehavior(
                                split[0],split[1],split[2],split[3],Long.parseLong(split[4])*1000L
                        ));
                    }
                })
                .filter(r -> r.behaviorType.equals("pv"))
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<UserBehavior>forMonotonousTimestamps()
                                .withTimestampAssigner(new SerializableTimestampAssigner<UserBehavior>(){
                                    @Override
                                    public long extractTimestamp(UserBehavior element, long recordTimestamp) {
                                        return element.timestamp;
                                    }
                                })
                )
                .keyBy(r -> true)
                .window(TumblingEventTimeWindows.of(Time.hours(1)))
                .aggregate(new Agg(),new WindowResult())
                .print();

        env.execute();
    }
    public static class Agg implements AggregateFunction< UserBehavior, Tuple2<BloomFilter<String>,Long>, Long>{

        @Override
        public Tuple2<BloomFilter<String>, Long> createAccumulator() {
            // 初始化累加器 bloom filter用户数量约为10万级别，误判率为0.01
            return Tuple2.of(BloomFilter.create(Funnels.stringFunnel(Charsets.UTF_8), 100000, 0.01),0L);
        }

        @Override
        public Tuple2<BloomFilter<String>, Long> add(UserBehavior value, Tuple2<BloomFilter<String>, Long> accumulator) {
            //如果过滤器中没有包含这个用户id,这个判断不是精确的,这是bloom filter 的特点
            if (!accumulator.f0.mightContain(value.userId)){
                //向bloom filter中添加元素
                accumulator.f0.put(value.userId);
                //累加结果+1
                accumulator.f1+=1L;
            }
            return accumulator;
        }

        @Override
        public Long getResult(Tuple2<BloomFilter<String>, Long> accumulator) {
            return accumulator.f1;
        }

        @Override
        public Tuple2<BloomFilter<String>, Long> merge(Tuple2<BloomFilter<String>, Long> a, Tuple2<BloomFilter<String>, Long> b) {
            return null;
        }
    }

    public static class WindowResult extends ProcessWindowFunction<Long,String,Boolean, TimeWindow>{
        @Override
        public void process(Boolean aBoolean, Context context, Iterable<Long> elements, Collector<String> out) throws Exception {
            out.collect(
                    "窗口" + new Timestamp(context.window().getStart())
                            + "~~" + new Timestamp(context.window().getEnd())
                            + "的uv数据是：" + elements.iterator().next()
            );
        }
    }

    public static class UserBehavior {
        public String userId;
        public String itemId;
        public String categoryId;
        public String behaviorType;
        public Long timestamp;

        public UserBehavior() {
        }

        public UserBehavior(String userId, String itemId, String categoryId, String behaviorType, Long timestamp) {
            this.userId = userId;
            this.itemId = itemId;
            this.categoryId = categoryId;
            this.behaviorType = behaviorType;
            this.timestamp = timestamp;
        }

        @Override
        public String toString() {
            return "UserBehavior{" +
                    "userId='" + userId + '\'' +
                    ", itemId='" + itemId + '\'' +
                    ", categoryId='" + categoryId + '\'' +
                    ", behaviorType='" + behaviorType + '\'' +
                    ", timestamp=" + new Timestamp(timestamp) +
                    '}';
        }
    }
}
